package com.atbeijing.app

import java.text.SimpleDateFormat
import java.util.Date
import com.alibaba.fastjson.JSON
import com.atbeijing.bean.StartUpLog
import com.atbeijing.constants.GmallConstants
import com.atbeijing.handler.DauHandler
import com.atbeijing.utils.MyKafkaUtil
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.phoenix.spark._

/**
 * 求日活,即每日登陆的用户数量
 */
object DauApp {
  def main(args: Array[String]): Unit = {
    //创建SparkConf
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("gmall")
    //创建StreamingContext,5秒一个批次
    val ssc = new StreamingContext(conf,Seconds(5))

    //当执行行动算子时,driver端获取一个批次要消费的kafka主题的分区的offset范围,一个kafka分区对应一个rdd分区,一个rdd分区对应一个task
    //当提交job时Driver将task(offset范围和计算逻辑)发给executor执行,executor从kafka拉取数据流进行计算
    val kafkaStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_STARTUP,ssc)

    //将数据流中json格式的数据转换为样例类
    //SimpleDateFormat已经实现了序列化接口,能在算子中使用
    val sdf = new SimpleDateFormat("yyyy-MM-dd HH")
    //rdd每个分区的数据应该怎么做,这里有两个分区,就有两个线程
    val startUpLogDStream: DStream[StartUpLog] = kafkaStream.mapPartitions(partition => {
      partition.map(record => {
        //将每条记录解析为一个StartUpLog对象
        val startUpLog: StartUpLog = JSON.parseObject(record.value(), classOf[StartUpLog])
        //格式化startUpLog中的ts,来补充样例类缺少的logDate和logHour
        val sdfDate: String = sdf.format(new Date(startUpLog.ts))
        val dateArray: Array[String] = sdfDate.split(" ")
        startUpLog.logDate = dateArray(0)
        startUpLog.logHour = dateArray(1)
        startUpLog
      })
    })
    //缓存在executor
    startUpLogDStream.cache()
    startUpLogDStream.count().print()

    //是先进行跨批次去重还是先进行批次内去重,取决于哪个取去掉的数据更多
    //跨批次去重,利用redis
    val filterByRedisDStream: DStream[StartUpLog] = DauHandler.filterByRedis(startUpLogDStream,ssc)
    filterByRedisDStream.cache()
    filterByRedisDStream.count().print()

    //批次内去重
    val myGroupByKey: DStream[StartUpLog] = DauHandler.myGroupByKey(filterByRedisDStream)
    myGroupByKey.cache()
    myGroupByKey.count().print()

    //将去重完成后的数据写入redis以便以后批次间去重
    DauHandler.writeRedis(myGroupByKey)

    //将数据写入hbase
    myGroupByKey.foreachRDD(rdd => {
      rdd.saveToPhoenix(
        "GMALL2021_DAU",
        Seq("MID", "UID", "APPID", "AREA", "OS", "CH", "TYPE", "VS", "LOGDATE", "LOGHOUR", "TS"),
        HBaseConfiguration.create,
        Some("hadoop202,hadoop203,hadoop204:2181"))
    })


    //程序持续运行
    ssc.start()
    ssc.awaitTermination()
  }
}
